package com.yeyks.rocketmq.service.customer;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.google.common.collect.Lists;
import com.yeyks.common.dto.mq.GrabTheOrderBToCMessage;
import com.yeyks.common.dto.mq.GrabTheOrderCToBMessage;
import com.yeyks.common.dto.mq.WebToBCnMessage;
import com.yeyks.common.em.WebSocketTypeEnum;
import com.yeyks.common.exception.ServiceException;
import com.yeyks.common.result.OrderPickWebSocket;
import com.yeyks.common.utils.ObjectConvertUtil;
import com.yeyks.commonReference.append.rocketmq.pojo.RocketMQWebsocketBasePojo;
import com.yeyks.commonReference.append.rocketmq.service.producer.MQProducerService;
import com.yeyks.commonReference.service.order.OrderPickRecordService;
import com.yeyks.commonReference.service.order.OrderPubRecordService;
import com.yeyks.order.dal.domain.OrderPickRecord;
import com.yeyks.order.dal.domain.OrderPubRecord;
import com.yeyks.search.pojo.param.EsC2bHotelSearchParam;
import com.yeyks.search.pojo.result.hotel.EsHotelInfoVo;
import com.yeyks.search.pojo.result.hotel.EsRoomInfoVo;
import com.yeyks.search.service.MySearchService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author fuchangshuai
 * @date 2019/08/09 16:06
 */
@Slf4j
@Component
public class CToWebOrderListener implements MessageListener {
    @Autowired
    private OrderPubRecordService orderPubRecordService;
    @Autowired
    private OrderPickRecordService orderPickRecordService;
    @Autowired
    private MQProducerService mqProducerService;
    @Autowired
    private MySearchService mySearchService;

    @Override
    public Action consume(Message message, ConsumeContext consumeContext) {
        try {
            Long startTime = System.currentTimeMillis();
            byte[] body = message.getBody();
            //获取到接收的消息，由于接收到的是byte数组，所以需要转换成字符串
            String msg = new String(body);
            log.info("抢单：监听到的消息：message = {}", msg);
            //处理业务逻辑
            disposeLogic(msg);
            Long endTime = System.currentTimeMillis();
            log.info("抢单：单次消费耗时：" + (endTime - startTime) / 1000);
        } catch (Exception e) {
            log.error("MessageListener.consume error:" + e.getMessage(), e);
        }

        log.info("MessageListener.Receive message");
        //如果想测试消息重投的功能,可以将Action.CommitMessage 替换成Action.ReconsumeLater
        return Action.CommitMessage;
    }

    /**
     * 3.4手动自动抢单，处理业务逻辑
     *
     * @param msg 监听到的信息
     */
    private void disposeLogic(String msg) {
        try {
            GrabTheOrderCToBMessage message = JSONUtil.toBean(msg, GrabTheOrderCToBMessage.class);
            log.info("监听后转换为对象: message = {}", message);
            //获取酒店搜索条件
            OrderPubRecord pubRecord = orderPubRecordService.getById(message.getOrderPubRecordId());
            if (null == pubRecord) {
                log.info("通过监听到的抢单记录表id查询到的记录为空:id = {}", message.getOrderPubRecordId());
                //发送mq告知搜索条件有误
                GrabTheOrderBToCMessage errorMessage = new GrabTheOrderBToCMessage(4, null, null, null);
                mqProducerService.webToCSendSimpleMsg(JSONUtil.toJsonStr(errorMessage));
                return;
            }
            //修改抢单状态为已抢单
            updateOrderPubRecord(pubRecord);
            //拼装搜索条件
            EsC2bHotelSearchParam param = ObjectConvertUtil.convertObject(pubRecord, EsC2bHotelSearchParam.class);
            if (null == param) {
                GrabTheOrderBToCMessage errorMessage = new GrabTheOrderBToCMessage(5, pubRecord.getUserId(), pubRecord.getTimestamp(), null);
                mqProducerService.webToCSendSimpleMsg(JSONUtil.toJsonStr(errorMessage));
                return;
            }
            //搜索自动抢单的酒店(5公里10公里20公里不少于15家)，自动抢单的直接写入数据(15家)
            param.setManualPickOrder(0);
            List<EsHotelInfoVo> autoEsHotelInfoVos = mySearchService.c2bSearchStepHotel(param);
            //过滤cn的酒店给b端发送mq
            filterCnToBMq(autoEsHotelInfoVos);
            autoInsertOrderPickRecord(autoEsHotelInfoVos, pubRecord);
            //搜索手动抢单酒店
//            param.setManualPickOrder(1);
//            param.setSourceType("yyks");
//            List<EsHotelInfoVo> noAutoEsHotelInfoVos = mySearchService.c2bSearchStepHotel(param);
//            if (noAutoEsHotelInfoVos.size() == 0) {
//                GrabTheOrderBToCMessage errorMessage = new GrabTheOrderBToCMessage(2, pubRecord.getUserId(), pubRecord.getTimestamp(), null);
//                mqProducerService.webToCSendSimpleMsg(JSONUtil.toJsonStr(errorMessage));
//                return;
//            }
//            //首先通过websocket给手动抢单的酒店推送消息，酒店点击抢单后将记录写入，加锁加判断是否存在
//            webSocketPush(noAutoEsHotelInfoVos, pubRecord);
        } catch (ServiceException e) {
            log.error("抢单：处理监听消息自定义异常：msg = {}", e.getMessage());
        } catch (Exception e) {
            e.printStackTrace();
            log.error("抢单：处理监听消息异常：msg = {}", msg);
        }
    }

    /**
     * 过滤cn的酒店给b端发送mq
     *
     * @param autoEsHotelInfoVos
     */
    private void filterCnToBMq(List<EsHotelInfoVo> autoEsHotelInfoVos) {
        log.info("给B端发mq执行-------begin-------");
        if (CollectionUtil.isEmpty(autoEsHotelInfoVos)) {
            return;
        }
        List<EsHotelInfoVo> collect = autoEsHotelInfoVos.stream().filter(e -> "cn".equals(e.getSourceType())).collect(Collectors.toList());
        if (CollectionUtil.isEmpty(collect)) {
            log.info("给B端发mq执行ing,过滤cn酒店结果为空");
            return;
        }
        List<WebToBCnMessage> webToBCnMessages = ObjectConvertUtil.convertList(collect, WebToBCnMessage.class);
        mqProducerService.webToBCnSendMsg(webToBCnMessages);
        log.info("给B端发送mq实现cn酒店更新实现hotels = {}", webToBCnMessages);
    }

    /**
     * 更新抢单状态为已抢单
     *
     * @param orderPubRecord
     */
    private void updateOrderPubRecord(OrderPubRecord orderPubRecord) {
        orderPubRecord.setStatus(1);
        orderPubRecordService.updateById(orderPubRecord);
    }

    /**
     * websocket推送手动抢单
     *
     * @param manualLists
     * @param pubRecord
     */
    private void webSocketPush(List<EsHotelInfoVo> manualLists, OrderPubRecord pubRecord) {
        List<OrderPickWebSocket> webSockets = Lists.newArrayList();
        if (null == manualLists) {
            return;
        }
        try {
            manualLists.forEach(e -> {
                OrderPickWebSocket webSocket = new OrderPickWebSocket();
                webSocket.setCheckIn(pubRecord.getStartDate().getTime());
                webSocket.setCheckOut(pubRecord.getEndDate().getTime());
                webSocket.setHotelId(e.getId());
                webSocket.setManualPickOrder(1);
                webSocket.setUserShowPrice(pubRecord.getPrice());
                webSocket.setPhone(pubRecord.getPhone());
                webSocket.setOperatorId(e.getOperatorId());
                webSocket.setUserId(pubRecord.getUserId());
                webSocket.setTimestamp(pubRecord.getTimestamp());
                webSocket.setSourceType(e.getSourceType());
                webSocket.setSoureHotelId(e.getSourceId());
                List<Integer> roomIds = Lists.newArrayList();
                for (EsRoomInfoVo roomInfoVo : e.getEsRoomInfoVos()) {
                    roomIds.add(roomInfoVo.getId());
                }
                webSocket.setRoomIds(roomIds);
                webSockets.add(webSocket);

            });
            if (webSockets.size() > 0) {
                webSockets.forEach(e -> {
                    RocketMQWebsocketBasePojo pojo = new RocketMQWebsocketBasePojo();
                    pojo.setType(WebSocketTypeEnum.ROB_ORDER.value());
                    pojo.setUserId(e.getOperatorId());
                    pojo.setBody(JSONUtil.toJsonStr(e));
                    mqProducerService.sendWebsocketMsg(pojo);
                    //发送mq让B端监听
                    mqProducerService.sendRobToBMsg(pojo);
                });
            }
        } catch (Exception exception) {
            log.error("websocket推送手动接单用户异常 pubRecord = {}", pubRecord);
            exception.printStackTrace();
        }
    }

    /**
     * 自动抢单入库
     *
     * @param autoLists
     * @param pubRecord
     */
    private void autoInsertOrderPickRecord(List<EsHotelInfoVo> autoLists, OrderPubRecord pubRecord) {
        List<OrderPickRecord> orderPickRecords = Lists.newArrayList();
        if (null == autoLists) {
            return;
        }
        autoLists.forEach(e -> {
            try {
                if (e.getEsRoomInfoVos().size() == 1) {
                    OrderPickRecord pickRecord = getOrderPickRecord(pubRecord, e);
                    pickRecord.setRoomId(e.getEsRoomInfoVos().get(0).getId());
                    if (StrUtil.isBlank(e.getSourceType())) {
                        log.info("酒店的sourceType为null,esHotelInfo = {}", e);
                    }
                    pickRecord.setSourceType(e.getSourceType());
                    pickRecord.setSoureHotelId(e.getSourceId());
                    pickRecord.setSourceRoomId(e.getEsRoomInfoVos().get(0).getSourceId());
                    orderPickRecords.add(pickRecord);
                } else {
                    for (EsRoomInfoVo roomInfoVo : e.getEsRoomInfoVos()) {
                        OrderPickRecord pickRecord = getOrderPickRecord(pubRecord, e);
                        if (StrUtil.isBlank(e.getSourceType())) {
                            log.info("酒店的sourceType为null,esHotelInfo = {}", e);
                        }
                        pickRecord.setSourceType(e.getSourceType());
                        pickRecord.setSoureHotelId(e.getSourceId());
                        pickRecord.setRoomId(roomInfoVo.getId());
                        pickRecord.setSourceRoomId(roomInfoVo.getSourceId());
                        orderPickRecords.add(pickRecord);
                    }
                }
            } catch (Exception exception) {
                log.error("自动接单对象处理异常 pubRecord = {}", pubRecord);
                exception.printStackTrace();
            }
        });
        if (orderPickRecords.size() > 0) {
            boolean saveBatch = orderPickRecordService.saveBatch(orderPickRecords);
            if (saveBatch) {
                GrabTheOrderBToCMessage errorMessage = new GrabTheOrderBToCMessage(11, pubRecord.getUserId(), pubRecord.getTimestamp(), null);
                mqProducerService.webToCSendSimpleMsg(JSONUtil.toJsonStr(errorMessage));
            }
        }
    }

    private OrderPickRecord getOrderPickRecord(OrderPubRecord pubRecord, EsHotelInfoVo e) {
        Date date = new Date();
        OrderPickRecord pickRecord = new OrderPickRecord();
        pickRecord.setCheckIn(pubRecord.getStartDate());
        pickRecord.setCheckOut(pubRecord.getEndDate());
        pickRecord.setGmtCreate(date);
        pickRecord.setGmtModified(date);
        pickRecord.setManualPickOrder(0);
        pickRecord.setUserShowPrice(pubRecord.getPrice());
        pickRecord.setUserId(pubRecord.getUserId());
        pickRecord.setPhone(pubRecord.getPhone());
        pickRecord.setTimestamp(pubRecord.getTimestamp());
        pickRecord.setOperatorId(e.getOperatorId());
        pickRecord.setHotelId(e.getId());
        pickRecord.setManualPickOrder(0);
        return pickRecord;
    }

}
